package com.fwmagic.spark.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 【【按照Key进行聚合，可以输入两个函数，第一个函数局部聚合，第二个函数全局聚合，
  *   初始值只在第一个函数聚合时使用，全局聚合不使用】】
  *
  * 对PairRDD中相同Key的值进行聚合操作，在聚合过程中同样使用了一个中立的初始值,
  * 因为aggregateByKey是对相同Key中的值进行聚合操作，所以aggregateByKey函数最终返回的类型还是Pair RDD，
  * 对应的结果是Key和聚合好的值
  */
object AggregateByKeyDemo {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
                .setAppName(this.getClass.getSimpleName)
                .setMaster("local[*]")

        val sc: SparkContext = new SparkContext(conf)

        val rdd: RDD[(String, Int)] = sc.parallelize(List(("cat",2),("cat",5),("mouse",4),
            ("cat",12),("dog",12),("mouse",2)),2)

        //初始值，局部聚合，全局聚合
        val res: RDD[(String, Int)] = rdd.aggregateByKey(0)(_+_,_+_)

        res.collect().foreach(println)

        sc.stop()

    }

}
